package flinkdemo.faceSink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

public class ImageClassSink extends RichSinkFunction<IdLabel> {
    private int index;
    private transient LogTimestampPlugin plugin = null;
    private String pluginClassName = System.getenv("LOG_PLUGIN_CLASSNAME");
    private String logPath;

    public ImageClassSink() {
        if (null == this.pluginClassName) {
            this.pluginClassName = PrintLogPlugin.class.getName();
        }

        this.logPath = System.getenv("SCORE_LOG_PATH");
        if (null == this.logPath) {
            this.logPath = "";
        }

        System.out.println(String.format("Plugin %s %s", this.pluginClassName, this.logPath));
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println(String.format("Plugin %s %s", this.pluginClassName, this.logPath));
        this.index = this.getRuntimeContext().getIndexOfThisSubtask();
        if (this.pluginClassName.isEmpty()) {
            this.plugin = null;
        } else {
            this.plugin = (LogTimestampPlugin)Class.forName(this.pluginClassName).newInstance();
        }

        if (null != this.plugin) {
            this.plugin.open(this.logPath, "sink", this.index);
        }

    }

    @Override
    public void close() throws Exception {
        if (null != this.plugin) {
            this.plugin.close();
        }

    }

    @Override
    public void invoke(IdLabel value, Context context) throws Exception {
        if (null != this.plugin) {
            this.plugin.logTimestamp(value.getId(), System.currentTimeMillis(), value.getLabel());
        }

    }
}